package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.bw.bean.OrderWide;
import com.bw.bean.YK8Result;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;


//jar->mysql->topic-db->kafka/redis->dwd_trade_orders ->  clickhouse
/*
create table yk8 (
                                    stt DateTime,
                                    edt DateTime,

                                    sku_id  String,
                                    sku_name String,
                                    province_id String,
                                    province_name String,
                                    user_id String,

                                    order_sku_num UInt64,
                                    order_amount Decimal64(2),
                                    order_ct UInt64,

                                    ts UInt64

)engine =ReplacingMergeTree( ts)
     order by   (stt,edt,sku_id );
 */
public class FlinkTM4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "group1");
        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_trade_orders_yk1", "test1112"));

        stream.print("input>>>>");

        //1）、编写Flink流式程序，从Kafka对象实时消费DWD层交易域下单事务事实表数据，设置允许乱序最大水位线为1秒，提取相关业务字段值，封装Java实体类对象；（5分）
        SingleOutputStreamOperator<OrderWide> mapDS = stream.map(x -> JSON.parseObject(x, OrderWide.class)).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner((event, timestamp) -> {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            try {
                return sdf.parse(event.getOrderInfo().getCreate_time()).getTime();
            } catch (ParseException e) {
                throw new RuntimeException(e);
            }
        }));
        //2）、设置事件时间窗口为1天，每隔1秒触发窗口计算，实时累加统计各个维度的指标度量中（订单数据、订单金额、商品数据量）；（5分）  skuid skuname provinceid provincename user_id
        //订单明细  订单   sku_num    splitTotalamount
        //1       1      2
        //2       1     2
        //3       1     2

        SingleOutputStreamOperator<YK8Result> resultDS1 = mapDS.map(new MapFunction<OrderWide, YK8Result>() {
            @Override
            public YK8Result map(OrderWide orderWide) throws Exception {
                return new YK8Result(null, null, orderWide.getOrderDetail().getSku_id(), orderWide.getOrderDetail().getSku_name(), orderWide.getBaseProvince().getId(), orderWide.getBaseProvince().getName(), orderWide.getUserInfo().getId(), 1L, Long.parseLong(orderWide.getOrderDetail().getSku_num()), Double.parseDouble(orderWide.getOrderInfo().getTotal_amount()), orderWide.getTs(),null);
            }
        });

        SingleOutputStreamOperator<YK8Result> xt2 = resultDS1.keyBy(new KeySelector<YK8Result, Tuple5<String, String, String, String, String>>() {
            @Override
            public Tuple5<String, String, String, String, String> getKey(YK8Result x) throws Exception {
                return new Tuple5<String, String, String, String, String>(x.getSku_id(), x.getSku_name(), x.getProvince_id(), x.getProvince_name(), x.getUser_id());
            }
        }).window(SlidingEventTimeWindows.of(Time.days(1), Time.minutes(1))).reduce(new ReduceFunction<YK8Result>() {
            @Override
            public YK8Result reduce(YK8Result yk8Result, YK8Result t1) throws Exception {
                yk8Result.setOrder_ct(yk8Result.getOrder_ct() + t1.getOrder_ct());
                yk8Result.setOrder_sku_num(yk8Result.getOrder_sku_num() + t1.getOrder_sku_num());
                yk8Result.setOrder_amount(yk8Result.getOrder_amount() + t1.getOrder_amount());
                return yk8Result;
            }
        }, new WindowFunction<YK8Result, YK8Result, Tuple5<String, String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple5<String, String, String, String, String> stringStringStringStringStringTuple5, TimeWindow timeWindow, Iterable<YK8Result> iterable, Collector<YK8Result> collector) throws Exception {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String start = sdf.format(timeWindow.getStart());
                String end = sdf.format(timeWindow.getEnd());

                for (YK8Result yk8Result : iterable) {
                    yk8Result.setStt(start);
                    yk8Result.setEdt(end);
                    yk8Result.setCur_date(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                    collector.collect(yk8Result);
                }

            }
        });

//        SingleOutputStreamOperator<YK8Result> xt2 = mapDS
//                .keyBy(new KeySelector<OrderWide, Tuple5<String, String, String, String, String>>() {
//                    @Override
//                    public Tuple5<String, String, String, String, String> getKey(OrderWide x) throws Exception {
//                        return new Tuple5<String, String, String, String, String>(
//                                x.getSkuInfo().getId(), x.getSkuInfo().getSku_name(),
//                                x.getBaseProvince().getId(), x.getBaseProvince().getName(),
//                                x.getUserInfo().getId());
//                    }
//                })
//                .window(SlidingEventTimeWindows.of(Time.days(1), Time.seconds(1)))
//                .process(new ProcessWindowFunction<OrderWide, YK8Result, Tuple5<String, String, String, String, String>, TimeWindow>() {
//                    @Override
//                    public void process(Tuple5<String, String, String, String, String> key, ProcessWindowFunction<OrderWide, YK8Result, Tuple5<String, String, String, String, String>, TimeWindow>.Context context, Iterable<OrderWide> iterable, Collector<YK8Result> collector) throws Exception {
//                        //订单数  count(1)
//                        //订单金额 sum(splitTotalamount)
//                        //商品数  sum(sku_num)
//                        Long orderCt = 0l;
//                        Long skuCt = 0l;
//                        Double amount = 0.0;
//                        for (OrderWide ow : iterable) {
//                            orderCt = orderCt + 1;
//                            skuCt = skuCt + Long.valueOf(ow.getOrderDetail().getSku_num().trim());
//                            amount = amount + Double.valueOf(ow.getOrderDetail().getSplit_total_amount().trim());
//                        }
//                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                        YK8Result r = new YK8Result(sdf.format(context.window().getStart()), sdf.format(context.window().getEnd()), key.f0, key.f1, key.f2, key.f3, key.f4, orderCt, skuCt, amount, new Date().getTime());
//                        collector.collect(r);
//                    }
//                });

        xt2.print();
        //3)、将上述交易域SKU粒度下单各窗口汇总结果数据，实时存储Clickhouse表，其中表的引擎ReplacingMergeTree主键相同时，更新字段值；（2分）
//        xt2.addSink(new DwsClickhouseSink());

        xt2.addSink(ClickHouseUtil.sink("insert into yk1 (stt,edt,sku_id,sku_name,province_id,province_name,user_id,order_ct,order_sku_num,order_amount,ts,cur_date) values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
